package cn.pink.dbsrv;

import cn.hutool.core.collection.CollectionUtil;
import cn.pink.common.constant.DBConst;
import cn.pink.common.constant.RedisConst;
import cn.pink.core.NodeType;
import cn.pink.core.Port;
import cn.pink.core.Service;
import cn.pink.core.config.IronConfig;
import cn.pink.core.gen.proxy.DistrClass;
import cn.pink.core.gen.proxy.DistrMethod;
import cn.pink.core.support.Log;
import cn.pink.core.support.Param;
import io.vertx.core.Future;
import io.vertx.core.Promise;
import io.vertx.core.json.JsonObject;
import io.vertx.ext.mongo.MongoClient;
import io.vertx.ext.mongo.UpdateOptions;
import io.vertx.redis.client.*;

import java.util.*;

/**
 * @Author: pink
 * @Date: 2022/11/16 17:46
 */
@DistrClass(nodeType = NodeType.DB, importClass = {Map.class, Set.class})
public class DBService extends Service {
    private MongoClient mongoClient;
    private RedisConnection redisClient;

    public DBService(Port port) {
        super(port);
        initMongoClient();
        initRedisClient();
    }

    @Override
    public Object getId() {
        return RPCProxy.DBServiceProxy.DIST_NAME;
    }

    private void initMongoClient() {
        JsonObject config = new JsonObject().put("connection_string", IronConfig.DB_MONGO_STR);
        mongoClient = MongoClient.create(getPort().getVertx(), config);
    }

    private void initRedisClient() {
        RedisOptions options = new RedisOptions();
        options.setConnectionString(IronConfig.DB_REDIS_STR);
        options.setMaxPoolSize(8);
        options.setMaxPoolWaiting(16);
        options.setMaxWaitingHandlers(128 * 1024);

        Redis.createClient(getPort().getVertx(), options).connect().onSuccess(conn -> {
            redisClient = conn;
            Log.db.info("conn redis success: connStr={}", IronConfig.DB_REDIS_STR);
        }).onFailure(Throwable::printStackTrace);
    }

    @DistrMethod
    public Param defaultHandler(String str) {
        return null;
    }

    @DistrMethod(isReturnAsync = true)
    public Future<Param> findHuman(String id) {
        Promise<Param> promise = Promise.promise();

        mongoClient.findOne(DBConst.DB_KEY_HUMAN, new JsonObject().put(DBConst.DB_KEY_ID, id), null).onSuccess(v -> {
            promise.complete(new Param(v == null ? "" : v.toString()));
        }).onFailure(Throwable::printStackTrace);

        return promise.future();
    }

    @DistrMethod
    public Param update(String humanId, Map<String, String> update, Set<String> unsets) {
        JsonObject queryObj = new JsonObject().put(DBConst.DB_KEY_ID, humanId);

        if(CollectionUtil.isNotEmpty(update)) {
            JsonObject updateObj = new JsonObject().put("$set", JsonObject.mapFrom(update));

            UpdateOptions options = new UpdateOptions().setUpsert(true);

            mongoClient.updateCollectionWithOptions(DBConst.DB_KEY_HUMAN, queryObj, updateObj, options);
        }

        if(CollectionUtil.isNotEmpty(unsets)) {
            JsonObject removeObjs = new JsonObject();
            unsets.forEach(v -> removeObjs.put(v, ""));

            JsonObject unsetObj = new JsonObject().put("$unset", removeObjs);

            mongoClient.updateCollection(DBConst.DB_KEY_HUMAN, queryObj, unsetObj);
        }

        return null;
    }

    @DistrMethod(isReturnAsync = true)
    public Future<Param> idAllot(long max) {
        Promise<Param> promise = Promise.promise();

        incrby(redisClient, RedisConst.DB_REDIS_KEY_ID_ALLOT, String.valueOf(max)).onSuccess(v -> {
            long idMin = v - max + 1L;
            promise.complete(new Param().put("idMin", idMin).put("idMax", v));
        });

        return promise.future();
    }

    @DistrMethod
    public Param cachePersist(String region, String key, byte[] data) {
        String rKey = RedisConst.DB_REDIS_KEY_CACHE + ":" + region;

        hset(redisClient, rKey, key, data);

        return null;
    }

    @DistrMethod(isReturnAsync = true)
    public Future<Param> cacheLoad(String region) {
        Promise<Param> promise = Promise.promise();

        String rKey = RedisConst.DB_REDIS_KEY_CACHE + ":" + region;

        hgetallBinary(redisClient, rKey).onSuccess(v -> {
            promise.complete(new Param(v));
        });

        return promise.future();
    }

    public Future<Long> incrby(RedisConnection conn, String key, String value) {
        Promise<Long> promise = Promise.promise();

        RedisAPI api = RedisAPI.api(conn);
        api.incrby(key, value).onSuccess(v -> promise.complete(v.toLong())).onFailure(Throwable::printStackTrace);

        return promise.future();
    }

    public void hset(RedisConnection conn, String key, String field, String value) {
        List<String> p = new ArrayList<>();
        p.add(key);
        p.add(field);
        p.add(value);

        RedisAPI api = RedisAPI.api(conn);
        api.hset(p).onFailure(Throwable::printStackTrace);
    }

    public void hset(RedisConnection conn, String key, String field, byte[] value) {
        conn.send(Request.cmd(Command.HSET).arg(key).arg(field).arg(value)).onFailure(Throwable::printStackTrace);
    }

    public Future<Map<String, String>> hgetall(RedisConnection conn, String region) {
        Promise<Map<String, String>> promise = Promise.promise();

        RedisAPI api = RedisAPI.api(conn);
        api.hgetall(region).onSuccess(v -> {
            Map<String, String> m = new HashMap<>();

            Iterator<Response> iterator = v.iterator();
            while(iterator.hasNext()) {
                m.put(iterator.next().toString(), iterator.next().toString());
            }

            promise.complete(m);
        }).onFailure(Throwable::printStackTrace);

        return promise.future();
    }

    public Future<Map<String, byte[]>> hgetallBinary(RedisConnection conn, String region) {
        Promise<Map<String, byte[]>> promise = Promise.promise();

        RedisAPI api = RedisAPI.api(conn);
        api.hgetall(region).onSuccess(v -> {
            Map<String, byte[]> m = new HashMap<>();

            Iterator<Response> iterator = v.iterator();
            while(iterator.hasNext()) {
                m.put(iterator.next().toString(), iterator.next().toBytes());
            }

            promise.complete(m);
        }).onFailure(Throwable::printStackTrace);

        return promise.future();
    }
}
